subscription_thread_dispatcher: better callback/thread lifecycle - #195
stephen-derosa wants to merge 8 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR refines SubscriptionThreadDispatcher’s reader-thread lifecycle handling to avoid redundant restarts and to ensure data reader threads reliably tear down (especially when a data-track subscription is still in flight), aligning with the SDK’s threading model by preventing teardown hangs.
Changes:
- Add cancellation signaling for active data readers and ensure cancellation is set before closing/joining during unpublish and teardown.
- Introduce a self-cleanup path (
eraseDataReaderIfCurrent) so data reader threads can safely remove stale slots when they exit. - Skip redundant audio/video/data reader restarts when the same track SID is already active.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.
| File | Description |
|---|---|
| src/tests/unit/test_subscription_thread_dispatcher.cpp | Adds unit tests covering data reader cancellation semantics and self-erase behavior. |
| src/subscription_thread_dispatcher.cpp | Implements cancellation-before-close, self-erase cleanup, and redundant reader-start suppression. |
| include/livekit/subscription_thread_dispatcher.h | Adds per-reader state (track_sid, cancelled) and declares the new cleanup helper. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
2f313fe to
f1ff6e6
Compare
| /// @return @c true if the callback was registered; @c false if a reader is | ||
| /// already active for the key (call @ref clearOnAudioFrameCallback | ||
| /// first) or the room has no dispatcher. | ||
| [[nodiscard]] bool trySetOnAudioFrameCallback(const std::string& participant_identity, const std::string& track_name, |
There was a problem hiding this comment.
From what I can tell, these new "try" methods have the same inputs as the deprecated prior ones, just a new return type (and name).
This is actually a situation where I think maintaining the API and using exceptions is cleaner than deprecation/new API just to get the bool return type.
Thoughts on this? Could we not move the new implementations from room.cpp into the older versions and throw instead of returning false?
There was a problem hiding this comment.
isnt that still changing the public API though? maybe im wrong but i would think adding a throw to a function that previously didnt would require a major bump?
There was a problem hiding this comment.
Fair point, technically speaking those methods were never marked noexcept and anything within them could throw (STL container exceptions, runtime exceptions from FFI stuff, etc.) so I think it's the least of the evils?
There was a problem hiding this comment.
hm yeah im a little torn here. I dont want to just start throwing underneath applications, especially when our mantra has somewhat been team #nothrow 😅
but the point of other things can throw inside the function are valid 🤔
There was a problem hiding this comment.
Also torn, I don't love my suggestion, to me keeping the API clean/avoiding deprecations is the least of the evils. Maybe @xianshijing-lk can chime in
| /// @return @c true if the callback was registered; @c false if a reader is | ||
| /// already active for the key (call @ref clearOnVideoFrameCallback | ||
| /// first) or the room has no dispatcher. | ||
| [[nodiscard]] bool trySetOnVideoFrameCallback(const std::string& participant_identity, const std::string& track_name, |
There was a problem hiding this comment.
I guess I didn't understand why these callback events need to be in the room ?
couldn't they be in the remote participants?
And does this frame callbacks work on both local / remote participants ?
There was a problem hiding this comment.
the room has access to the room events which are needed to set callbacks to incoming track messages -- the remote participant doesnt really.
This callback only works for local participants, the intention is to provide a nice ergonomic way for the user to set callbacks for incoming messages.
| } else { | ||
| // The track is not subscribed yet. The callback is registered; the reader | ||
| // starts when the track is subscribed (see kTrackSubscribed in onEvent). | ||
| LK_LOG_DEBUG( |
There was a problem hiding this comment.
should you still return true here ?
There was a problem hiding this comment.
good question, yes since the user has successfully set the callback. It doesnt necessarily mean that a track has been published or that they will get messages on said track. If we were to only return true to a callback being set after a track has been published, then early joining participants that want incoming messages will effectively have to periodically set callbacks until success, or listen to room events -- both of which defeat the purpose of the ergonomic functionality of these setOn*Callback functions.
| // If we've already subscribed to the track, handle it immediately | ||
| auto track = findSubscribedRemoteTrack(participant_identity, track_name); | ||
| if (track) { | ||
| subscription_thread_dispatcher_->handleTrackSubscribed(participant_identity, track_name, track); |
There was a problem hiding this comment.
I think I forgot some technical details here, is this subscription_thread_dispatcher_ a manager that manages all the subscription thread ?
and does one callback has one corresponding subscription thread ? or one thread that handles all the callbacks ?
There was a problem hiding this comment.
I think I forgot some technical details here, is this subscription_thread_dispatcher_ a manager that manages all the subscription thread ?
it manages subscribing to tracks and providing incoming frames to the users callbacks
and does one callback has one corresponding subscription thread ? or one thread that handles all the callbacks ?
Each subscription gets its own thread
2179d4c to
fbdcbd4
Compare
|
ensure that this also works for local publishing of video/audio |
4f18b6f to
2aa6d4c
Compare
4b75ba8 to
0dbdc64
Compare
ab8df3a to
bb9224e
Compare
…plicate room events test SubscriptionThreadDispatcher: proper replacing of audio/video callbacks. Deprecate a setOn*Callback(), replace with trySetOn*Callback() fix thread detaching
27e9cae to
f5d5055
Compare
There was a problem hiding this comment.
Devin Review found 3 new potential issues.
7 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| /// Currently subscribed remote audio/video tracks keyed by @ref CallbackKey. | ||
| std::unordered_map<CallbackKey, std::shared_ptr<Track>, CallbackKeyHash> subscribed_tracks_; | ||
|
|
||
| /// Keys whose previous reader has been extracted but not yet joined, with the | ||
| /// number of such in-progress drains. No reader is started for a key while | ||
| /// it has an entry here. See @ref extractReaderForDrainLocked. | ||
| std::unordered_map<CallbackKey, int, CallbackKeyHash> draining_readers_; |
There was a problem hiding this comment.
🔴 Dispatcher ABI corrupts existing clients
Existing clients allocate the old SubscriptionThreadDispatcher size, but its constructor now initializes a larger layout. The added maps write beyond caller-allocated storage, causing memory corruption or crashes.
Learn more
SubscriptionThreadDispatcher is a public, exported class with an out-of-line constructor. C++ clients allocate its storage using the size recorded when they compiled. Adding data members changes that size without changing the constructor symbol, so an old executable can load the new shared library and call the new constructor on an undersized object. The repository's public ABI rule requires backwards compatibility.
Example: An application compiled against the previous release embeds SubscriptionThreadDispatcher dispatcher;. After replacing only liblivekit.so, the new constructor initializes subscribed_tracks_ and draining_readers_ beyond the application's allocated object.
Recommended fix: Preserve the released class layout. Move evolving dispatcher state behind an ABI-stable private implementation pointer, or defer this layout change to a declared ABI-breaking major release.
Was this helpful? React with 👍 or 👎 to provide feedback.
| old_thread = extractReaderForDrainLocked(key); | ||
| const bool replacing = audio_callbacks_.find(key) != audio_callbacks_.end(); | ||
| audio_callbacks_[key] = RegisteredAudioCallback{std::move(callback), opts}; | ||
| LK_LOG_DEBUG( | ||
| "Registered audio frame callback for participant={} track_name={} " | ||
| "replacing_existing={} stopped_reader={} total_audio_callbacks={}", | ||
| participant_identity, track_name, replacing, old_thread.joinable(), audio_callbacks_.size()); | ||
| } | ||
| // Joins the previous reader first, then starts a fresh one bound to the new | ||
| // callback if the track is subscribed. | ||
| finishReaderDrainAndRestart(key, std::move(old_thread), "setOnAudioFrameCallback"); |
There was a problem hiding this comment.
🟡 Concurrent disconnect leaves callback running
Concurrent disconnect() misses a reader extracted by setOn*FrameCallback while that setter waits for its callback. Disconnect returns before the callback ends, so teardown does not quiesce frame delivery.
Learn more
Replacement removes the active reader from active_readers_ before joining it. The setter then owns the only std::thread handle while the callback can remain in flight. A concurrent stopAll scans only active_readers_, so it cannot close or join this draining reader and can return first. The dispatcher is documented as thread-safe, and stopAll() is the shutdown path that must stop all readers.
Example: Callback A blocks for ten seconds. Thread 1 replaces A and waits in join(). Thread 2 calls Room::disconnect(); stopAll() sees no active reader and returns while A keeps running.
Recommended fix: Make shutdown coordinate with in-progress drains. Track drain completion with shared reader state or a condition variable, and make stopAll() wait for every non-self draining reader without holding lock_.
Was this helpful? React with 👍 or 👎 to provide feedback.
| subscribed_tracks_.erase(key); | ||
| old_thread = extractReaderForDrainLocked(key); |
There was a problem hiding this comment.
🟡 Stale unsubscribe stops new publication
A delayed unsubscribe makes handleTrackUnsubscribed erase the current same-name subscription without comparing SIDs. The new publication's reader stops and receives no restart event.
Learn more
The dispatcher retains media subscriptions by participant identity and track name, but a subscription event also identifies a concrete publication by SID. Room event forwarding resolves the unsubscribe SID, then discards it and passes only the publication name. If an old unsubscribe is delivered after a same-name republish or resubscribe, these lines erase and stop whichever SID is currently retained.
Example: Alice's camera publication TR_old is replaced by TR_new. After TR_new starts its reader, a delayed unsubscribe for TR_old reaches the dispatcher and removes TR_new, leaving the callback silent.
Recommended fix: Pass the unsubscribed SID into handleTrackUnsubscribed. Under lock_, erase and drain only when the retained track and active reader still match that SID.
Was this helpful? React with 👍 or 👎 to provide feedback.
| std::unordered_map<CallbackKey, ActiveReader, CallbackKeyHash> active_readers_; | ||
|
|
||
| /// Currently subscribed remote audio/video tracks keyed by @ref CallbackKey. | ||
| std::unordered_map<CallbackKey, std::shared_ptr<Track>, CallbackKeyHash> subscribed_tracks_; |
There was a problem hiding this comment.
will this create races or confusion to our users regarding the subscribed tracks ?
I think our room has API about RemoteTrackPublication, that tracks whether a publication is subscribed. And this |subscribed_tracks_| might confuse developers on which one is the ground truth regarding the subscribed tracks ?
or are they strictly in sync?
Overview
Fixes #235 and hardens reader-thread lifecycle in
SubscriptionThreadDispatcher. Frame callbacks now start a reader regardless of when they are registered, replacement actually replaces, and driving the dispatcher from inside a frame callback no longer self-joins. No public API changes —setOn*FrameCallbacknow does what it always documented.The bugs
handleTrackSubscribedran once, right afterRoomDelegate::onTrackSubscribed, and no-oped without a registered callback. Registering afterwards (e.g. from a GUI thread) stored the callback and nothing else happened.Room::disconnect()orremoveOnDataFrameCallbackcalled from inside a data frame callback joined the calling thread:std::system_error, then a still-joinablestd::threaddestroyed during unwinding.Changes
(participant, track_name)until unsubscribed, sosetOn*starts a reader immediately when the track is already subscribed and defers otherwise.Roomalso resolves the current publication, which covers registration from insideonTrackSubscribed.setOn*,clearOn*, resubscribe with a new SID, unsubscribe) extracts the reader under the lock, marks the key draining, joins outside the lock, then restarts. No caller can start a reader for a draining key, so old and new callbacks never run concurrently and concurrent setters cannot pile up untracked threads.track_subscribedfor the SID a reader already serves is a no-op; a new SID (republish) stops the old reader first.this; data readers report exit through their own shared state. All readers are therefore safe to detach, so a lifecycle call from inside the reader's own callback logs a warning, detaches, and still takes effect.RemoteDataTrack::subscribe()calls are marked cancelled so unpublish/republish cannot leave a stale reader or duplicate FFI subscription behind.SubscriptionThreadDispatcherdeprecated for direct use (LIVEKIT_DEPRECATED); it stays exported to avoid a major bump.Behavioral notes
setOn*/clearOn*block until any in-flight invocation of the previous callback returns. A callback that never returns blocks registration indefinitely.Public API
No signature changes.
include/livekit/room.hchanges are documentation only. Callback type aliases moved toinclude/livekit/frame_callbacks.h(still reachable throughroom.h).Test-only:
RemoteDataTrack::testFfiHandleId()removed in favor ofRemoteDataTrackTestAccess;RoomTestAccessexposes reader/drain counts and retained-subscription state;RemoteDataTrack's private constructor isLIVEKIT_INTERNAL_APIso tests can link it.Testing
SubscriptionThreadDispatcherTest, 91): Calling setOnVideo/AudioFrameCallback outside of the onTrackSubscribed handler never starts a frame thread handler #235 in both orderings for audio/video/video-event; replacement semantics including the shared video slot; SID dedup and republish; drain marks set and cleared by every lifecycle path, with starts deferred while draining; self-join handling forsetOn*,clearOn*,removeOnDataFrameCallback, andstopAllfrom both media and data reader threads; data reader cancellation and finished-state handling.FrameCallbackServerTest, 5): the exact Calling setOnVideo/AudioFrameCallback outside of the onTrackSubscribed handler never starts a frame thread handler #235 scenario for video, video-event, and audio; clear-and-re-register without a new event; registration from insideonTrackSubscribedstarting exactly one reader.FrameCallbackReplacementTest, 14): delivery switches on replacement; setter blocks for a slow callback without overlapping it; sequential and concurrent churn leave exactly one reader; deferred start and republish; re-entrantsetOn*/clearOn*/data removal take effect;Room::disconnect()from inside a data callback.